package com.chukun.netty.nio;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @author chukun
 * java原生nio客户端
 */
public class NIOClient {

    private static ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("java nio client thread pool").build();

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 5, 20, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10),threadFactory, new ThreadPoolExecutor.AbortPolicy());

    public static void main(String[] args) {
        try {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
            socketChannel.connect(new InetSocketAddress(8899));
            Selector selector = Selector.open();
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
            while(true) {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = iterator.next();
                    if (selectionKey.isConnectable()) {
                        SocketChannel sc = (SocketChannel) selectionKey.channel();
                        if (sc.isConnectionPending()) {
                            sc.finishConnect();
                            ByteBuffer writtenBuf = ByteBuffer.allocate(1024);
                            writtenBuf.put((LocalDateTime.now() + " connected success").getBytes());
                            writtenBuf.flip();
                            sc.write(writtenBuf);
                            threadPoolExecutor.submit(() -> {
                                while (true) {
                                    BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
                                    try {
                                        writtenBuf.clear();
                                        String line = br.readLine();
                                        writtenBuf.put(line.getBytes("utf-8"));
                                        writtenBuf.flip();
                                        sc.write(writtenBuf);
                                    } catch (IOException e) {
                                        e.printStackTrace();
                                    }
                                }
                            });
                        }
                        //注册读事件
                        sc.register(selector, SelectionKey.OP_READ);
                    } else if (selectionKey.isReadable()) {
                        SocketChannel sc = (SocketChannel) selectionKey.channel();
                        ByteBuffer readBuf = ByteBuffer.allocate(1024);
                        int read = sc.read(readBuf);
                        if (read > 0) {
                            System.out.println(new String(readBuf.array(), 0, read, "utf-8"));
                        }
                    }
                    //去除监听的key
                    iterator.remove();
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
